package org.apache.rocketmq.proxy.grpc;

import io.grpc.BindableService;
import io.grpc.ServerInterceptor;
import io.grpc.ServerServiceDefinition;
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
import io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup;
import io.grpc.netty.shaded.io.netty.channel.epoll.EpollServerSocketChannel;
import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup;
import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioServerSocketChannel;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.grpc.interceptor.ContextInterceptor;
import org.apache.rocketmq.proxy.grpc.interceptor.GlobalExceptionInterceptor;
import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor;

import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class GrpcServerBuilder {

    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);


    //Netty服务器的构建器，Grpc底层也是使用Netty作为通信框架
    protected NettyServerBuilder serverBuilder;

    //服务器默认的关闭时间
    protected long time = 30;

    //定义时间单位
    protected TimeUnit unit = TimeUnit.SECONDS;

    public GrpcServerBuilder(ThreadPoolExecutor executor, int port) {
        this.serverBuilder = NettyServerBuilder.forPort(port);
        //设置协议协商器，用来处理代理和TLS协议的协商过程
        serverBuilder.protocolNegotiator(new ProxyAndTlsProtocolNegotiator());
        //从配置信息对象中得到Grpc服务器要使用的Boss线程池的线程数
        int bossLoopNum = ConfigurationManager.getProxyConfig().getGrpcBossLoopNum();
        //从配置信息对象中得到Grpc服务器要使用的Worker线程池的线程数
        int workLoopNum = ConfigurationManager.getProxyConfig().getGrpcWorkerLoopNum();
        //从配置信息对象中得到Grpc服务器的最大入站消息大小
        int maxInboundMessageSize = ConfigurationManager.getProxyConfig().getGrpcMaxInboundMessageSize();
        //从配置信息对象中得到Grpc服务器的空闲时间
        long grpcIdleTimeout = ConfigurationManager.getProxyConfig().getGrpcClientIdleTimeMills();
        //判断是否要启用Epoll模式
        boolean enableEpoll = ConfigurationManager.getProxyConfig().isEnableGrpcEpoll();
        if (enableEpoll){
            //如果启用Epoll模式，使用Epoll模式配置Boss线程池和Worker线程池，以及Channel类型
            this.serverBuilder.bossEventLoopGroup(new EpollEventLoopGroup(bossLoopNum))
                    .workerEventLoopGroup(new EpollEventLoopGroup(workLoopNum))
                    .channelType(EpollServerSocketChannel.class)
                    .executor(executor);
        }else{
            this.serverBuilder.bossEventLoopGroup(new NioEventLoopGroup(bossLoopNum))
                    .workerEventLoopGroup(new NioEventLoopGroup(workLoopNum))
                    .channelType(NioServerSocketChannel.class)
                    .executor(executor);
        }
        this.serverBuilder.maxInboundMessageSize(maxInboundMessageSize).maxConnectionIdle(grpcIdleTimeout,unit);
        log.info(
                "grpc server has built. port: {}, tlsKeyPath: {}, tlsCertPath: {}, threadPool: {}, queueCapacity: {}, "
                        + "boosLoop: {}, workerLoop: {}, maxInboundMessageSize: {}",
                port, bossLoopNum, workLoopNum, maxInboundMessageSize);
    }

    public static GrpcServerBuilder newBuilder(ThreadPoolExecutor executor, int port) {
        return new GrpcServerBuilder(executor, port);
    }


    //添加服务到gRPC服务器的方法
    public GrpcServerBuilder addService(BindableService service) {
        this.serverBuilder.addService(service);
        return this;
    }

    //设置服务器关闭的超时时间
    public GrpcServerBuilder shutdownTime(long time, TimeUnit unit) {
        this.time = time;
        this.unit = unit;
        return this;
    }
    //添加服务定义信息对象到gRPC服务器的方法
    public GrpcServerBuilder addService(ServerServiceDefinition service) {
        this.serverBuilder.addService(service);
        return this;
    }

    //添加拦截器到gRPC服务器的方法
    public GrpcServerBuilder appendInterceptor(ServerInterceptor interceptor) {
        this.serverBuilder.intercept(interceptor);
        return this;
    }

    public GrpcServer build() {
        return new GrpcServer(this.serverBuilder.build(), time, unit);
    }

    //向Grpc服务器设置拦截器的方法
    public GrpcServerBuilder configInterceptor() {
        this.serverBuilder
                .intercept(new GlobalExceptionInterceptor())
                .intercept(new ContextInterceptor())
                .intercept(new HeaderInterceptor());
        return this;
    }
}